package my.thesis.bolts.fieldExtractors;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import my.thesis.analytics.metrics.ThroughputMetric;

import org.joda.time.DateTime;
import org.json.JSONArray;
import org.json.JSONObject;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class TweetFieldExtractorTopo2 extends BaseRichBolt 
{
		private static final long serialVersionUID = 1L;	
		private OutputCollector collector;
			
		transient ThroughputMetric _throughputMetric;
		private static Boolean metricsOn=false;

		public void prepare(Map stormConf, TopologyContext context,OutputCollector collector)
		{
			this.collector=collector;
			
			if(metricsOn)
			{
				_throughputMetric=new ThroughputMetric();
				context.registerMetric("fieldTopo2", _throughputMetric, 20);
				_throughputMetric.setStartTime();
			}
		}
			
		public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) 
		{		
				outputFieldsDeclarer.declareStream("textStream", new Fields("id","millis","country","lang","text","lat","lon"));   
				outputFieldsDeclarer.declareStream("influenceStream", new Fields("id","username","mention","followers","text")); 
		}
		
		public void execute(Tuple tuple) 
		{
			if(metricsOn)
			{
				_throughputMetric.incrTuples();
			}
			
			Set<String> mentions = new HashSet<String>();

			String jsonString = tuple.getStringByField("jsonString");
			
			JSONObject obj = null;
			try
			{
				obj = new JSONObject(jsonString);
			}
			catch(Exception ex)
			{
				System.out.println("Something went wrong...");
			}
			
			if(obj != null)
			{
				Long id = obj.getLong("id");
				String username=obj.getString("username");
				int followers = obj.getInt("followers");
				
				String lang = obj.getString("lang");
				
				String text = obj.getString("text");
				String country = obj.getString("country");
							
				DateTime timestamp=new DateTime(obj.get("createdat"));
				Long millis = timestamp.getMillis();
				
				//TEST
				//DateTime timestamp = DateTime.now();
				//Long millis = timestamp.getMillis();
				//TEST
				
				Double  lat = obj.getDouble("lat");
				Double lon = obj.getDouble("lon");
								
				if(obj.has("mentions"))
				{
					JSONArray mentionsArray = obj.getJSONArray("mentions");
					for (int i=0; i < mentionsArray.length(); i++)
					{
						mentions.add(mentionsArray.getJSONObject(i).getString("name"));
					}
				}
				
				collector.emit("textStream", new Values(id,millis,country,lang,text,lat,lon));
				
				for (String mention : mentions)
				{
					collector.emit("influenceStream",new Values(id,username,mention,followers,text));
				}				
			}
			else
			{
				//do nothing
			}		
		}		
}
